package cn.gwm.flink.streaming.strategy.vehiclepredict;

import cn.gwm.flink.streaming.constant.ChargePredictFields;
import cn.gwm.utils.HBaseDataSource;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.math.BigDecimal;
import java.sql.PreparedStatement;
import java.sql.SQLException;

/**
 * @author : GW00280745
 * @date : 2023/4/26 13:05
 * @description :
 * 实时更新每1%soc的时间到phoenix中
 *
 */
public class OneSocSink extends RichSinkFunction<JSONObject> {

    @Override
    public void invoke(JSONObject line, Context context) throws Exception {
        try {
            String vinVal = line.getString(ChargePredictFields.ins().srcFieldVin);
            Long tidVal = line.getLong(ChargePredictFields.ins().srcFieldTid);
            String startChargeTime = line.getString(ChargePredictFields.ins().srcFieldStartChargeTime);
            Float bmsSoc = line.getFloat(ChargePredictFields.ins().srcFieldBmsSoc);
            //先判断状态是否有数据
            boolean contains = oneSocTimeState.contains(vinVal);
            //没有，则state中的值是0
            String socState = null;
            if (!contains){
                socState = tidVal.toString().concat(ChargePredictFields.ins().char3).concat(bmsSoc.toString());
                oneSocTimeState.put(vinVal,socState);
            }else {
                //有，则进行判断 soc差值是否为整数
                socState = oneSocTimeState.get(vinVal);
            }
            float delta = deltaSoc(bmsSoc, socState);
            //判断 soc差值是否为整数 找大于0.9的第一个值
            boolean b = delta >= 1.0 ;
            if (!b){
                return;
            }
            //符合条件则计算时间差
            String[] split = socState.split(ChargePredictFields.ins().char3);
            String timeState = split[0];
            long l = Long.parseLong(timeState);
            long deltaTime = tidVal - l;
            float v = new BigDecimal(deltaTime).divide(new BigDecimal(60 * 1000), 0, BigDecimal.ROUND_HALF_DOWN).floatValue();
            data2Phoenix(line,v);
        }catch (Exception exception){
            exception.printStackTrace();
        }
    }
    private float deltaSoc(Float bmsSoc, String socState) {
        String[] split = socState.split(ChargePredictFields.ins().char3);
        String s = split[1];
        float v = Float.parseFloat(s);
        float delta = bmsSoc - v;
        return delta;
    }
    private java.sql.Connection connection;
    private PreparedStatement preparedStatement;
    private MapState<String, String> oneSocTimeState;
    @Override
    public void open(Configuration parameters) throws Exception {
        try {
            String sql = "upsert into \"vaas_dws\".\"dws_multidimension_charge_obc_v3\"(id,\"bms_innersoc\",\"startchargetime\",\"bms_rmc_moduletempmin\",\"e\",\"model_code\",\"bms_battcurr_rm\",\"poiid\",\"aoiid\") values(?,?,?,?,?,?,?,?,?)";
            connection = HBaseDataSource.ins().getConnection();
            connection.setAutoCommit(false);
            preparedStatement = connection.prepareStatement(sql);
            oneSocTimeState = getRuntimeContext().getMapState(new MapStateDescriptor<String, String>("oneSocTimeState", String.class, String.class));
        }catch (Exception e){
            e.printStackTrace();
        }
    }
    @Override
    public void close() throws Exception {
        if (cntNum>0){
            exeSql();
        }
        if (connection!=null){
            connection.close();
        }
        if (preparedStatement!=null){
            preparedStatement.close();
        }
    }
    private int maxNum = 300;
    private int cntNum = 0;
    private void data2Phoenix(JSONObject line, float v) throws Exception{
        sqlConcat(line, v, preparedStatement);
        preparedStatement.addBatch();
        cntNum+=1;
        if (cntNum>=maxNum){
            exeSql();
        }
    }

    private void exeSql() throws SQLException {
        long start = System.currentTimeMillis();
        int[] ints = preparedStatement.executeBatch();
        connection.commit();
        cntNum=0;
        preparedStatement.clearBatch();
        long end = System.currentTimeMillis();
        System.out.println("phoenix sql 花费时间为(ms):" + (end - start));
    }

    private void sqlConcat(JSONObject line, float v, PreparedStatement preparedStatement) throws SQLException {
        String vinVal = line.getString(ChargePredictFields.ins().srcFieldVin);
        String tidVal = line.getString(ChargePredictFields.ins().srcFieldTid);
        String startChargeTime = line.getString(ChargePredictFields.ins().srcFieldStartChargeTime);
        String bmsSoc = line.getString(ChargePredictFields.ins().srcFieldBmsSoc);
        String temp = line.getString(ChargePredictFields.ins().srcFieldBmsRmcModuleTempMin);
        String modelCode = line.getString(ChargePredictFields.ins().srcFieldModelCode);
        String curr = line.getString(ChargePredictFields.ins().srcFieldBmsBatterCurrRm);
        String poiId = line.getString(ChargePredictFields.ins().srcFieldPoiId);
        String aoiId = line.getString(ChargePredictFields.ins().srcFieldAoiId);

        String id = vinVal.concat(kangStr).concat(StringUtils.reverse(tidVal));
//        id = "test123456"
        preparedStatement.setString(1,id);
        String bmsSocConvert = socConvert(bmsSoc);
        preparedStatement.setString(2,bmsSocConvert);
        preparedStatement.setString(3,startChargeTime);
        String tempConvert = tempConvert(temp);
        preparedStatement.setString(4,tempConvert);
        preparedStatement.setString(5,String.valueOf(v));
        preparedStatement.setString(6,modelCode);
        String currConvert = currConvert(curr);
        preparedStatement.setString(7,currConvert);
        preparedStatement.setString(8,poiId);
        preparedStatement.setString(9,aoiId);
    }
    private String kangStr= "-";
    private int num1=1;
    private int num2=2;
    private int num3=3;
    private int num4=4;
    private int num10=10;
    private String str100="100";
    private String str100of0="100.0";

    private String socConvert(String soc){
        int length = soc.length();
        if (length==num1){
            return  "00".concat(soc).concat(".0");
        }else if (length == num2){
            return  "0".concat(soc).concat(".0");
        }else if (length == num3){
            if (StringUtils.equalsIgnoreCase(soc,str100) || StringUtils.equalsIgnoreCase(soc,str100of0)){
                return "100.0";
            }else {
                return "00".concat(soc);
            }
        }else if (length == num4){
            return  "0".concat(soc);
        }
        return null;
    }
    private String tempConvert(String tempMin){
        float v = Float.parseFloat(tempMin);
        float abs = Math.abs(v);

        if (v<0){
            if (abs<num10){
                return "-0".concat(String.valueOf((int)abs));
            }else {
                return tempMin;
            }
        }else if (v>0){
            if (abs<num10){
                return "a0".concat(String.valueOf((int)abs));
            }else {
                return "a".concat(String.valueOf((int)abs));
            }
        }else {
            return "a00";
        }
    }
    private String currConvert(String curr){
        float v = Float.parseFloat(curr);
        float abs = Math.abs(v);
        if (abs<num10){
            return "-0".concat(String.valueOf(abs));
        }
        return curr;
    }
}
